/**
 * 事件定时器
 */


package com.atguigu.day4.keyed

import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.sql.Timestamp

object ProcessingTimeTimer {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(1)
    //
    val stream1:DataStream[String] = env.socketTextStream("192.168.35.102", 9999, '\n')
    stream1.keyBy(t=>t)
            .process(new Keyed)
              .print()
    env.execute()
  }

  class Keyed extends KeyedProcessFunction[String,String,String]{
    override def processElement(value: String, ctx: KeyedProcessFunction[String, String, String]#Context, out: Collector[String]): Unit = {
      ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime())

    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, String, String]#OnTimerContext, out: Collector[String]): Unit = {

      out.collect("定时器执行的时间戳为"+ new Timestamp(timestamp))
    }
  }
}